<?php

namespace console\db\channels;

use Yii;
use yii\db\Query;
use yii\helpers\Console;
use yii\helpers\ArrayHelper;
use yii\base\Exception;
use console\db\Migration;

/**
 * 数据表直接复制迁移
 * 
 * @author emhome <emhome@163.com>
 * @since 1.0
 */
abstract class TransferTableMigration extends Migration {

    /**
     * 数据分页量级限制
     */
    const PART_MAX = 100000;

    /**
     * 批次查询数据条数
     */
    const PART_SIZE = 30000;

    /**
     * 分表大小
     */
    const MERGE_SIZE = 0;

    /**
     * 预留空间大小
     * @inheritdoc
     */
    const TABLE_RESERVE_ROW_SIZE = 10000;

    /**
     * 分表索引
     */
    public $partIndex = false;

    /**
     * @var bool 是否先清理本地数据表
     */
    public $beforeTruncateTargetTable = true;

    /**
     * @var string 源数据库
     */
    public $sourceDb = null;

    /**
     * @var string 目标数据库
     */
    public $targetDb = null;

    /**
     * @var string 数据源表
     */
    private $_sourceTable;

    /**
     * @var string 目标表
     */
    private $_targetTable;

    /**
     * @inheritdoc
     */
    public function init() {
        parent::init();
        if ($this->db->driverName === 'mysql') {
            $this->tableOptions = 'CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci ENGINE=' . $this->tableEngine;
        }
        $table = ArrayHelper::getValue($this->getTransferOptions(), 'table');
        $this->setTransferTable($table);
    }

    /**
     * 源数据库连接
     * @return \console\db\Connection 源数据库连接
     */
    public function getSourceDb() {
        if ($this->sourceDb !== null) {
            return Yii::$app->{$this->sourceDb};
        }
        return $this->db;
    }

    /**
     * 目标数据库连接
     * @return \console\db\Connection 源数据库连接
     */
    public function getTargetDb() {
        if ($this->targetDb !== null) {
            return Yii::$app->{$this->targetDb};
        }
        return $this->db;
    }

    /**
     * 自定义查询
     * @return \yii\db\Query|null
     */
    public function buildQuery() {
        return null;
    }

    /**
     * 目标表
     * @return string
     */
    public function getTargetTable($merge = false) {
        if ($merge !== false && is_numeric($merge)) {
            return $this->_targetTable . '_' . $merge;
        }
        return $this->_targetTable;
    }

    /**
     * 目标表
     * @return string
     */
    public function getSourceTable() {
        return $this->_sourceTable;
    }

    /**
     * 初始化转移数据表
     */
    public function setTransferTable($table = []) {
        if (!empty($table) && is_array($table)) {
            foreach ($table as $target => $source) {
                if (is_numeric($target) && is_string($source)) {
                    $target = $source;
                }
                $this->_targetTable = $target;
                $this->_sourceTable = $source;
                break;
            }
        }
    }

    /**
     * 获取转移数据表结构字段
     * @return array 
     */
    public function getTransferTableColumns($fields = []) {
        if (!is_array($fields) || empty($fields)) {
            return [];
        }
        $columns = [];
        foreach ($fields as $key => $vls) {
            if (is_string($key)) {
                $columns[$key] = $vls;
            } elseif (is_numeric($key) && is_string($vls)) {
                $columns[$vls] = $vls;
            } else {
                throw new Exception('错误');
            }
        }
        return $columns;
    }

    /**
     * 预留目标表空间
     * @inheritdoc
     */
    public function reserveRowsTable($tableName) {
        $table = $this->basenameTable($tableName, true);
        $now = $this->getTableAutoIncrementId($table);
        $base = static::TABLE_RESERVE_ROW_SIZE;
        $incs = ceil($now / $base) * $base + 1;

        $time = $this->beginCommand("Reserve table $table's rows");
        $sql = "ALTER TABLE {$table} AUTO_INCREMENT=" . ($incs) . ";";
        $this->db->createCommand($sql)->execute();
        $this->endCommand($time);
    }

    /**
     * 目标查询字段，默认*
     * @inheritdoc
     */
    public function filterQuerySelects() {
        return [];
    }

    /**
     * 目标查询字段，默认*
     * @inheritdoc
     */
    public function filterQueryConditions($targetTable, $sourceTable) {
        $tableBatchRecord = $this->getBatchIncrementTable($targetTable);
        if (empty($tableBatchRecord)) {
            return [];
        }
        $column = ArrayHelper::getValue($tableBatchRecord, 'increment_column');
        $start = ArrayHelper::getValue($tableBatchRecord, 'increment_start');
        $end = ArrayHelper::getValue($tableBatchRecord, 'increment_end');
//        if ($start && $end) {
//            $this->beforeTruncateTargetTable = false;
//        }
        if (!$end || !$column) {
            return [];
        }
        return [
            'and',
            ['>=', $column, $start],
            ['<', $column, $end],
        ];
    }

    /**
     * @inheritdoc
     */
    public function safeUp() {
        $options = $this->getTransferOptions();
        $fields = ArrayHelper::getValue($options, 'columns');
        $primaryKeys = ArrayHelper::getValue($options, 'primaryKey', []);
        if (!empty($primaryKeys)) {
            $sourcePrimaryKey = current($primaryKeys);
        } else {
            $sourcePrimaryKey = 'id';
        }
        $columns = $this->getTransferTableColumns($fields);

        $query = $this->buildQuery();
        if (!($query && $query instanceof Query)) {
            $query = (new Query)->from('{{%' . $this->getSourceTable() . '}}');
            $conditions = $this->filterQueryConditions($this->getTargetTable(), $this->getSourceTable());
            if (!empty($conditions)) {
                $query->andWhere($conditions);
            }
        }
        $selects = $this->filterQuerySelects();
        if (!empty($selects)) {
            $query->select($selects);
        }

        $db = $this->getSourceDb();
        $time = microtime(true);
        Console::output("Calculate the number of paging data: ");
        $total = $query->count('*', $db);
        if (!$total) {
            Console::output("No data transferred.");
            return;
        }
        Console::output("*** summary query times: " . sprintf('%.3f', microtime(true) - $time) . "s.");

        if ($this->beforeTruncateTargetTable) {
            $this->forceTruncateTable('{{%' . $this->getTargetTable() . '}}');
        }
        $callable = [$this, 'transferTable'];
        if ($total > static::PART_MAX) {
            Console::output("*** The current data has exceeded the order of magnitude limit, and efficient paging query processing is adopted.");
            $pagesize = static::PART_SIZE;
            $pages = ceil($total / $pagesize);
            $query->limit($pagesize);

            Console::startProgress(1, $pages, 'Updating objects for parts: ', true);

            if ($pages > 10) {
                for ($i = 0; $i < $pages; $i++) {
                    if (self::MERGE_SIZE) {
                        $this->partIndex = floor(($i * $pagesize) / static::MERGE_SIZE);
                    }
                    $query->offset($i * $pagesize);
                    $query->select($sourcePrimaryKey);

                    $subQuery = (new Query())->from(['sub' => $query])
                    ->innerJoin(['m' => $query->from[0]], 'm.' . $sourcePrimaryKey . '=sub.' . $sourcePrimaryKey)
                    ->select('m.*');

                    $this->handleProgress($subQuery, $callable, $columns, $db);
                    Console::updateProgress($i + 1, $pages);
                }
            } else {
                for ($i = 0; $i < $pages; $i++) {
                    $query->offset($i * $pagesize);
                    $this->handleProgress($query, $callable, $columns, $db);
                    Console::updateProgress($i + 1, $pages);
                }
            }

            Console::endProgress("big data update done.");
        } else {
            $this->handleProgress($query, $callable, $columns, $db, $total);
        }
        $this->afterTransfer($this->getTargetTable());
    }

    /**
     * @inheritdoc
     */
    public function safeDown() {
        $targetTable = $this->getTargetTable();
        $tableName = '{{%' . $targetTable . '}}';
        if ($this->beforeTruncateTargetTable && $this->isExistsTable($tableName)) {
            $this->truncateTable($tableName);
        }
    }

    /**
     * 表对表之间转移
     * 
     * For example
     * array structure
     * 
     * ```php
     * return [
     *     'table' => [
     *          'targetTableName' => 'sourceTableName'
     *      ],
     *     'columns' => [
     *          'targetTableColumnName' => 'sourceTableColumnName'
     *      ],
     * ];
     * ```
     * 
     * columns source supported callable function.
     * For example
     * 
     * ```php
     * return [
     *     ```
     *     'columns' => [
     *          'targetTableColumnName' => function($row){
     *              //$row is source table row data
     *              //transfer code
     *          },
     *      ],
     * ];
     * ```
     * 
     * @return array
     */
    abstract public function getTransferOptions();

    /**
     * 处理结果
     * @param Migration $migration
     * @param array $rows
     */
    public static function transferTable($migration, $rows, $fields) {
        $data = [];
        foreach ($rows as $i => $row) {
            if (!$migration->validateTransfer($row)) {
                continue;
            }
            $append = ArrayHelper::remove($row, '_append', []);
            $appends = ArrayHelper::remove($row, '_appends', []);
            $temp = $migration->convertColumnData($row, $fields);
            unset($rows[$i]);
            if (empty($temp)) {
                continue;
            }
            $data[] = $temp;
            if (!empty($append)) {
                $appends[] = $append;
            }
            if (!empty($appends)) {
                foreach ($appends as $append) {
                    $tempAppend = $migration->convertColumnData($append, $fields);
                    if (!empty($tempAppend)) {
                        $data[] = $tempAppend;
                    }
                }
            }
        }
        $targetTable = $migration->getTargetTable($migration->partIndex);
        $migration->db->createCommand()->batchInsert('{{%' . $targetTable . '}}', array_keys($fields), $data)->execute();
    }

    /**
     * @inheritdoc
     */
    public function convertColumnData($row, $fields) {
        $data = [];
        foreach ($fields as $name => $field) {
            if (is_string($field)) {
                $data[] = ArrayHelper::getValue($row, $field);
            } elseif (is_callable($field)) {
                $data[] = call_user_func($field, $row);
            } else {
                throw new Exception("field {$name} datatype invalid.");
            }
        }
        return $data;
    }

}
